package com.artisan.spider.kafka;

import com.alibaba.fastjson.JSON;
import com.artisan.spider.domain.SpLawChl;
import com.artisan.spider.service.ISpLawChlService;
import com.artisan.spider.util.DateUtil;
import com.artisan.spider.util.LuceneUtil;
import com.artisan.spider.util.ResourceUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
 * 消费队列中的数据
 */
@Component
public class KafkaConsumer {

    @Autowired
    ISpLawChlService iSpLawChlService;

    @KafkaListener(topics = "messageQueen")
    public void listen(ConsumerRecord<?, ?> record){

        //如果服务重启 以前积攒的信息会重复消费
        SpLawChl spLawChl = iSpLawChlService.getById(record.value().toString());
        if(spLawChl!=null&&!spLawChl.getIsCreate().equals("1")){
            // 创建索引库
            createIndex(spLawChl);
            // 更新表字段未已创建过
            spLawChl.setIsCreate("1");
            spLawChl.setUpdateTime(DateUtil.getCurrentTime());
            iSpLawChlService.updateById(spLawChl);
        }
    }

    private void createIndex(SpLawChl lawInfo) {
        String[] fields = new String[]{"id","pzDeptCode", "pzDeptName", "fbDeptName",
                "fbDeptCode", "lssuedNumber", "releaseDate", "timeName", "timeCode",
                "xiaoliName", "xiaoliCode", "faguiName", "faguiCode", "implDate","fgfl","isCreate"
        };
        String[] contents = new String[]{"fullText","title"};
        String lucenePath = ResourceUtil.getString("AUDIT_LUCENE_PATH");
        LuceneUtil lucene = new LuceneUtil("id", lucenePath, Arrays.asList(fields), Arrays.asList(contents));
        String lawInfoStrt = JSON.toJSONString(lawInfo);
        try {
            lucene.index(JSON.parseObject(lawInfoStrt));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
